package spark.sql

import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkContext, SparkConf}

/**
 * Created by ibf on 2018/2/10.
 */
object UDFandUDAFSpark {
  def main(args: Array[String]) {
    //1、构建spark上下文(当你构建的时候，判断下，时候之前构建过
    // 如果构建过，尝试能不能直接获取这个上下文，如果拿不到，
    // 考虑是否有getOrCreate的方法，实在获取不到，再创建
    // SparkContext在同一个jvm中创建是会出问题的
    // HiveContext或者SQLContext这个多次创建是不会报错的
    // 但是会消耗更多内存，也没有必要去多次创建)
    val conf = new SparkConf()
          .setMaster("local[*]")
          .setAppName("UDFandUDAFSpark")
    val sc = SparkContext.getOrCreate(conf)

    /**
     * 自定义UDF和UDAF，所以需要sqlcontext
     */
    val sqlContext = new SQLContext(sc)

    /**
     * udf和udaf和udtf的区别
     *    UDF:一条数据输入，对应一条数据输出：一般用在数据格式的转换
     *   UDAF:多条数据输入，对应一条数据输出：聚合函数，SUM,COUNT,AVG,MAX,MIN
     *   UDTF:一条数据输入，对应多条数据输出：ip的解析，从一个字段当中返回多个值出来
     *
     * 注：spark暂时只能实现udf和udaf，udtf实现不了
     */
    //构建保留小数点后两位的udf
    sqlContext.udf.register("format_double",(value:Double) =>{
      //这里就不自己手动实现，直接调用默认已经有的java类
      // 记得调用前要手动导入包
      import java.math.BigDecimal
      val bd = new BigDecimal(value)
      // bd.setScale(2,BigDecimal.ROUND_HALF_UP).doubleValue()是自动返回一个double类型的数据
      bd.setScale(2,BigDecimal.ROUND_HALF_UP).doubleValue()
    })

    sqlContext.udf.register("format_double_ano",(value:Double,num:Int) =>{
      //这里就不自己手动实现，直接调用默认已经有的java类
      // 记得调用前要手动导入包
      import java.math.BigDecimal
      val bd = new BigDecimal(value)
      // bd.setScale(2,BigDecimal.ROUND_HALF_UP).doubleValue()是自动返回一个double类型的数据
      bd.setScale(num,BigDecimal.ROUND_HALF_UP).doubleValue()
    })

    //自定义UDAF，AVG
    sqlContext.udf.register("self_avg",Avg_UDAF)


      //2、构建DataFrame数据
      // （当使用toDF的API的时候，要导入
      //  import sqlContext.implicits._）
    import sqlContext.implicits._
    sc.parallelize(Array(
      (1,1231),
      (1,1234),
      (1,43435),
      (1,23443),
      (2,34354),
      (2,3423),
      (2,123),
      (3,23423),
      (3,7743),
      (4,12312),
      (4,4545),
      (5,343),
      (5,23426)
    )).toDF("id","sal").registerTempTable("tmp_emp")

    sqlContext.sql(
      """
        |SELECT id,AVG(sal) as avg_sal1,
        |format_double(AVG(sal)) as avg_sal2,
        |format_double_ano(AVG(sal),1) as avg_sal3,
        |self_avg(sal) as avg_sal4,
        |format_double_ano(self_avg(sal),1) as avg_sal5
        |FROM tmp_emp
        |GROUP BY id
      """.stripMargin).show


  }

}
